import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.ArrayList;
import java.util.List;

public class SqlEngine {
    public static void main(String[] args) {

        System.out.println("Run SQL engine");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
        ParameterTool fromArgs = ParameterTool.fromArgs(args);
        // 传递sql
        String sql = fromArgs.get("sql");
        // 假设sql中只包含create和insert语句，并且没有注释
        List<String> sqls = SqlUtil.split(sql);
        List<String> createSql = new ArrayList<>();
        List<String> insertSql = new ArrayList<>();
        for (String s : sqls) {
            if(s.toLowerCase().startsWith("create")) {
                createSql.add(s);
            } else if (s.toLowerCase().startsWith("insert")) {
                insertSql.add(s);
            } else {
                System.out.println("不支持的SQL:" + sql);
            }
        }

        // 执行create sql
        for (String s : createSql) {
            tabEnv.executeSql(s);
        }

        // 将insert sql添加到StatementSet中执行
        StatementSet statementSet = tabEnv.createStatementSet();
        for (String s : insertSql) {
            statementSet.addInsertSql(s);
        }
        statementSet.execute();
    }
}
